package com.yitiao.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;

@RestController
@RequestMapping("/kafka")
@Slf4j
public class KafkaController {

    private final KafkaTemplate<String,String> kafkaTemplate;

    public KafkaController(KafkaTemplate<String,String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }


    @GetMapping("/send")
    public void send(@RequestParam(defaultValue = "hi")String msg) throws ExecutionException, InterruptedException {
        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send("quickstart-events", msg);
        String res = listenableFuture.get().toString();
        log.info("producer send result:[{}]",res);
    }

}
